package com.psw.monitor.geserialport.kafka;

import com.psw.monitor.geserialport.model.SerialPortMessage;
import com.psw.monitor.geserialport.utils.ConvertHexStrAndStrUtils;
import com.psw.monitor.geserialport.utils.MsgConverter;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.Future;

public class GEKafkaProducer {

    private static Logger logger = LoggerFactory.getLogger(GEKafkaProducer.class);

    private static KafkaProducer<String, String> producer;

    private static final String TOPIC = "geChannel";

    private static final String BOOTSTRAP_SERVERS_CONFIG = "127.0.0.1:9092";

    static {
        // 构建连接配置
        Properties properties = new Properties();

        // 若要配多个服务器，用逗号隔开
        // 注：服务器要开放端口，若云服务器还要在server.properties配置内网IP和外网IP
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS_CONFIG);
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "my-producer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        properties.put(ProducerConfig.ACKS_CONFIG,"0");
//        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,1);
        // 构造Client无非是：new 或 工厂模式
        producer = new KafkaProducer<String, String>(properties);
    }

    private GEKafkaProducer() {

    }

    public static void send(SerialPortMessage msg) {
        long now = System.currentTimeMillis();

        String systemId = ConvertHexStrAndStrUtils.bytesToHexStr(msg.getAddr());

        String contentStr = MsgConverter.fromMsg(msg);
        String topic = TOPIC;
        String msgType = "S1";


        if(msg.getHeader()[0]!=-86 || msg.getHeader()[1]!=-69 || msg.getFooter()[0]!=-52 || msg.getFooter()[1]!=-35){
            return;
        }
        contentStr = now + ":" + contentStr;
        logger.info("{}发送中:{}",systemId, contentStr);
        Future<RecordMetadata> f = producer.send(new ProducerRecord<String, String>(topic, systemId, systemId + ":" + contentStr));
        try {
            f.get();
            logger.info("发送成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
